<?php

namespace Diablo7\HyperfPulsar;

use Diablo7\HyperfPulsar\Exception\ConnectException;
use Hyperf\Contract\ContainerInterface;
use Hyperf\Pool\KeepaliveConnection;
use Hyperf\Pool\Pool;

class PulsarConnection extends KeepaliveConnection
{
    public $name = "pulsar.connection";

    /**
     * @var PulsarClient
     */
    public $client;

    public function __construct(ContainerInterface $container,Pool $pool,PulsarClient $client)
    {
        parent::__construct($container,$pool);

        $this->client = $client;
    }

    public function getActiveConnection()
    {
        if(!$this->client->connect())
        {
            throw new ConnectException(sprintf(
                'Websocket upgrade failed by [%s] [%s].',
                $this->client->errCode,
                swoole_strerror($this->client->errCode)
            ));
        }
        return $this;
    }

    protected function heartbeat(): void
    {
        $this->client->ping();
    }

    /**
     * Send close protocol.
     * @param mixed $connection
     */
    protected function sendClose($connection): void
    {
        $this->client->close();
    }
}